package integration.data.task.api.channel.impl;

import integration.data.task.api.channel.Channel;
import integration.modle.exception.IntegrationValidException;
import io.github.bigbird0101.datatransfer.record.Record;
import io.github.bigbird0101.datatransfer.record.impl.TerminateRecord;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.Collection;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class MemoryNonBlockingChannel extends Channel implements Serializable {
    private AtomicInteger memoryBytes = new AtomicInteger(0);
    private ConcurrentLinkedQueue<Record> queue;
    private ReentrantLock lock;

    public MemoryNonBlockingChannel() {
    }

    //    private Condition notInsufficient, notEmpty;
    public MemoryNonBlockingChannel(String config) {
        super(config);
        this.queue = new ConcurrentLinkedQueue<>();
        lock = new ReentrantLock();
//        notInsufficient = lock.newCondition();
//        notEmpty = lock.newCondition();
    }

    @Override
    public void close() {
        super.close();
        this.queue.offer(TerminateRecord.get());
    }

    @Override
    public void clear() {
        this.queue.clear();
    }

    @Override
    protected void doPush(Record r) {
        long startTime = System.nanoTime();
        this.queue.offer(r);
        waitWriterTime += System.nanoTime() - startTime;
        memoryBytes.addAndGet(r.getMemorySize());
    }

    @Override
    protected void doPushAll(Collection<Record> rs) {
        throw new IntegrationValidException("不支持批量push");
//        try {
//            long startTime = System.nanoTime();
//            lock.lockInterruptibly();
//            int bytes = rs.stream().mapToInt(Record::getMemorySize).sum();
//            while (memoryBytes.get() + bytes > this.byteCapacity || rs.size() > this.capacity-this.queue.size()) {
//                notInsufficient.await(200L, TimeUnit.MILLISECONDS);
//            }
//            this.queue.addAll(rs);
//            waitWriterTime += System.nanoTime() - startTime;
//            memoryBytes.addAndGet(bytes);
//            notEmpty.signalAll();
//        } catch (InterruptedException e) {
//            Thread.currentThread().interrupt();
//            throw new IntegrationValidException("Error during execution", e);
//        } finally {
//            lock.unlock();
//        }
    }

    @Override
    protected Record doPull() {
        try {
            long startTime = System.nanoTime();
            lock.lockInterruptibly();
            try {
                if (this.queue.isEmpty()) {
                    return TerminateRecord.get();
                }
                Record r = this.queue.poll();
                waitReaderTime += System.nanoTime() - startTime;
                memoryBytes.addAndGet(-r.getMemorySize());
                return r;
            } finally {
                lock.unlock();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IntegrationValidException("Error during execution", e);
        }
    }

    @Override
    protected void doPullAll(Collection<Record> rs) {
        throw new IntegrationValidException("不支持批量pull");
//        try {
//            long startTime = System.nanoTime();
//            lock.lockInterruptibly();
//            while (this.queue.isEmpty()) {
//                notEmpty.await();
//            }
//            rs.clear();
//
//            int size = this.queue.size();
//            for (int i = 0; i < size; i++) {
//                rs.add(this.queue.poll());
//            }
////            int n = this.queue.drainTo(rs);
//            waitReaderTime += System.nanoTime() - startTime;
//            int bytes = rs.stream().mapToInt(Record::getMemorySize).sum();
//            memoryBytes.addAndGet(-bytes);
//            notInsufficient.signalAll();
//        } catch (InterruptedException e) {
//            Thread.currentThread().interrupt();
//            throw new IntegrationValidException("Error during execution", e);
//        } finally {
//            lock.unlock();
//        }
    }

    @Override
    public int size() {
        return this.queue.size();
    }

    @Override
    public boolean isEmpty() {
        return this.queue.isEmpty();
    }
}
